package com.bole.reactivestream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author: 叶子
 * @date: 2022/04/10 18:48
 * 该demo需要jdk9.0以上
 **/
public class ReactiveStreamDemo {
    public static void main(String[] args) {
        //1.创建一个发布者
        SubmissionPublisher publisher = new SubmissionPublisher();
        //2.创建一个订阅者
        Flow.Subscriber subscriber = new Flow.Subscriber() {
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println("建立订阅关系");
                this.subscription = subscription;
                subscription.request(1);//第一次需要

            }

            @Override
            public void onNext(Object item) {
                System.out.println("接收数据：" + item);
                //业务处理
                subscription.request(10);//回压
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("发生错误了");
            }

            @Override
            public void onComplete() {
                System.out.println("数据接收完成");
            }
        };
        //3.建立一个订阅者
        publisher.subscribe(subscriber);
        for (int i = 0; i <100; i++) {
            //4.发送数据
            publisher.submit("hello reactive stream："+i);
        }
        try {
            Thread.currentThread().join();//reactive是异步流的处理，需要主线程阻塞
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }
}